package com.info33.platform.common.MQservice.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.info33.platform.common.MQservice.IMessageService;
import com.info33.platform.common.exception.BaseException;
import com.info33.platform.common.util.MQConstantUtil;
import com.info33.platform.common.util.MQMessageUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * 消息队列服务接口实现
 */
@Service("IMessageService")
@Slf4j
public class IMessageServiceimpl implements IMessageService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void send(MQMessageUtil message) throws BaseException {
        this.checkMessage(message);
        //即时消息
        if (message.getType() == 1) this.sendMessage(message.getExchange(), message.getQueueName(), message.getMessage());
        //延时消息
        if (message.getType() == 2) sendTimeMessage(message);
    }

    private void sendMessage(String exchange, String queueName, String msg) {
        rabbitTemplate.convertAndSend(exchange, queueName, msg);
    }

    public void sendTimeMessage(MQMessageUtil message) {
        int seconds = message.getSeconds();
        if (seconds <= 0) {// 直接发送，无需进入死信队列
            sendMessage(message.getExchange(), message.getQueueName(), message.getMessage());
        } else {
            long times = seconds * 1000;//rabbit默认为毫秒级
            MessagePostProcessor processor = s -> {
                s.getMessageProperties().setHeader("x-delay", times);
                return s;
            };
            //设置 x-delay 之后,将消息投递到专属交换机,转发队列
            rabbitTemplate.convertAndSend(MQConstantUtil.DEFAULT_DELAYED_EXCHANGE, MQConstantUtil.DEFAULT_REPEAT_TRADE_QUEUE_NAME, JSON.toJSONString(message), processor);
        }
    }


    private void checkMessage(MQMessageUtil message) throws BaseException {
        if (StrUtil.isEmpty(message.getExchange())) throw new BaseException("发送消息格式错误: 消息交换机(exchange)不能为空!");
        if (message.getGroup() == null) throw new BaseException("发送消息格式错误: 消息组(group)不能为空!");
        if (message.getType() == null) throw new BaseException("发送消息格式错误: 消息类型(type)不能为空!");
        if (message.getStatus() == null) throw new BaseException("发送消息格式错误: 消息状态(status)不能为空!");
        if (StrUtil.isEmpty(message.getQueueName())) throw new BaseException("发送消息格式错误: 消息目标名称(queueName)不能为空!");
        if (StrUtil.isEmpty(message.getMessage())) throw new BaseException("发送消息格式错误: 消息内容(message)不能为空!");
    }

}